package com.atguigu.day04;

import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Flink07_OrderWatch {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从文件读取数据并转为JavaBean是为了方便获取数据
        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new OrderEvent(Long.parseLong(split[0]), split[1], split[2], Long.parseLong(split[3]));
                    }
                });

        SingleOutputStreamOperator<TxEvent> txEventStream = env.readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new TxEvent(split[0], split[1], Long.parseLong(split[2]));
                    }

                });

        //3.使用Connect连接两条流
        ConnectedStreams<OrderEvent, TxEvent> connect = orderEventStream.connect(txEventStream);

        //TODO 4.对相同交易码的数据聚合到一块 （在多并行度的情况下，一定要将数据聚合到一块，如果并行度是1问题不大，先做KeyBy还是先做Connect效果一样
        ConnectedStreams<OrderEvent, TxEvent> orderEventTxEventConnectedStreams = connect.keyBy("txId", "txId");

        //5.关联两个流的数据把匹配到的打印到控制台  (在没有合适的算子能够实现需求的时候都可以用Process)
        SingleOutputStreamOperator<String> result = orderEventTxEventConnectedStreams.process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {

            //用来临时存放订单表数据
            private HashMap<String, OrderEvent> eventHashMap = new HashMap<>();

            //用来临时存放交易表数据
            private HashMap<String, TxEvent> txHashMap = new HashMap<>();

            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.查询交易表的集合中是否有能关联上的数据
                if (txHashMap.containsKey(value.getTxId())) {
                    //有能关联上的数据
                    out.collect("订单:" + value.getOrderId() + "关联成功！！！！！");
                    //删除已关联的数据
                    txHashMap.remove(value.getTxId());
                } else {
                    //没有能关联上的数据
                    //将自己存入集合中
                    eventHashMap.put(value.getTxId(), value);
                }

            }

            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.查询订单表的集合中是否有能关联上的数据
                if (eventHashMap.containsKey(value.getTxId())) {
                    //有能关联上的数据
                    out.collect("订单:" + eventHashMap.get(value.getTxId()).getOrderId() + "关联成功！！！！！");
                    //删除已关联的数据
                    eventHashMap.remove(value.getTxId());
                } else {
                    //没有能关联上的数据
                    //将自己存入集合中
                    txHashMap.put(value.getTxId(), value);
                }
            }
        });

        result.print();

        env.execute();
    }
}
